// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/compute/expression.h>

#include <algorithm>
#include <optional>
#include <unordered_map>
#include <unordered_set>

#include <nebula/core/chunked_array.h>
#include <nebula/compute/api_vector.h>
#include <nebula/compute/exec_internal.h>
#include <nebula/compute/expression_internal.h>
#include <nebula/compute/function_internal.h>
#include <nebula/compute/util.h>
#include <nebula/io/memory.h>
#include <nebula/ipc/reader.h>
#include <nebula/ipc/writer.h>
#include <nebula/util/hash_util.h>
#include <nebula/util/key_value_metadata.h>
#include <turbo/log/logging.h>
#include <nebula/util/string.h>
#include <nebula/util/value_parsing.h>
#include <nebula/util/vector.h>

namespace nebula::compute {

    using nebula::internal::ToChars;

    void Expression::Call::ComputeHash() {
        hash = std::hash<std::string>{}(function_name);
        for (const auto &arg: arguments) {
            nebula::internal::hash_combine(hash, arg.hash());
        }
    }

    Expression::Expression(Call call) {
        call.ComputeHash();
        impl_ = std::make_shared<Impl>(std::move(call));
    }

    Expression::Expression(Datum literal)
            : impl_(std::make_shared<Impl>(std::move(literal))) {}

    Expression::Expression(Parameter parameter)
            : impl_(std::make_shared<Impl>(std::move(parameter))) {}

    Expression literal(Datum lit) { return Expression(std::move(lit)); }

    Expression field_ref(FieldRef ref) {
        return Expression(Expression::Parameter{std::move(ref), TypeHolder{}, {-1}});
    }

    Expression call(std::string function, std::vector<Expression> arguments,
                    std::shared_ptr<compute::FunctionOptions> options) {
        Expression::Call call;
        call.function_name = std::move(function);
        call.arguments = std::move(arguments);
        call.options = std::move(options);
        return Expression(std::move(call));
    }

    const Datum *Expression::literal() const {
        if (impl_ == nullptr) return nullptr;

        return std::get_if<Datum>(impl_.get());
    }

    const Expression::Parameter *Expression::parameter() const {
        if (impl_ == nullptr) return nullptr;

        return std::get_if<Parameter>(impl_.get());
    }

    const FieldRef *Expression::field_ref() const {
        if (auto parameter = this->parameter()) {
            return &parameter->ref;
        }
        return nullptr;
    }

    const Expression::Call *Expression::call() const {
        if (impl_ == nullptr) return nullptr;

        return std::get_if<Call>(impl_.get());
    }

    const DataType *Expression::type() const {
        if (impl_ == nullptr) return nullptr;

        if (const Datum *lit = literal()) {
            return lit->type().get();
        }

        if (const Parameter *parameter = this->parameter()) {
            return parameter->type.type;
        }

        return CallNotNull(*this)->type.type;
    }

    namespace {

        std::string PrintDatum(const Datum &datum) {
            if (datum.is_scalar()) {
                if (!datum.scalar()->is_valid) return "null[" + datum.type()->to_string() + "]";

                switch (datum.type()->id()) {
                    case Type::STRING:
                    case Type::LARGE_STRING:
                        return '"' +
                               Escape(std::string_view(*datum.scalar_as<BaseBinaryScalar>().value)) + '"';

                    case Type::BINARY:
                    case Type::FIXED_SIZE_BINARY:
                    case Type::LARGE_BINARY:
                        return '"' + datum.scalar_as<BaseBinaryScalar>().value->ToHexString() + '"';

                    default:
                        break;
                }

                return datum.scalar()->to_string();
            } else if (datum.is_array()) {
                return "Array[" + datum.type()->to_string() + "]";
            }
            return datum.to_string();
        }

    }  // namespace

    std::string Expression::to_string() const {
        if (auto lit = literal()) {
            return PrintDatum(*lit);
        }

        if (auto ref = field_ref()) {
            if (auto name = ref->name()) {
                return *name;
            }
            if (auto path = ref->field_path()) {
                return path->to_string();
            }
            return ref->to_string();
        }

        auto call = CallNotNull(*this);
        auto binary = [&](std::string op) {
            return "(" + call->arguments[0].to_string() + " " + op + " " +
                   call->arguments[1].to_string() + ")";
        };

        if (auto cmp = Comparison::Get(call->function_name)) {
            return binary(Comparison::GetOp(*cmp));
        }

        constexpr std::string_view kleene = "_kleene";
        if (turbo::ends_with(call->function_name, kleene)) {
            auto op = call->function_name.substr(0, call->function_name.size() - kleene.size());
            return binary(std::move(op));
        }

        if (auto options = GetMakeStructOptions(*call)) {
            std::string out = "{";
            auto argument = call->arguments.begin();
            for (const auto &field_name: options->field_names) {
                out += field_name + "=" + argument++->to_string() + ", ";
            }
            out.resize(out.size() - 1);
            out.back() = '}';
            return out;
        }

        std::string out = call->function_name + "(";
        for (const auto &arg: call->arguments) {
            out += arg.to_string() + ", ";
        }

        if (call->options) {
            out += call->options->to_string();
        } else if (call->arguments.size()) {
            out.resize(out.size() - 2);
        }

        out += ')';
        return out;
    }

    void print_to(const Expression &expr, std::ostream *os) {
        *os << expr.to_string();
        if (expr.is_bound()) {
            *os << "[bound]";
        }
    }

    bool Expression::equals(const Expression &other) const {
        if (Identical(*this, other)) return true;

        if (impl_->index() != other.impl_->index()) {
            return false;
        }

        if (auto lit = literal()) {
            // The scalar NaN is not equal to the scalar NaN but the literal NaN
            // is equal to the literal NaN (e.g. the expressions are equal even if
            // the values are not)
            EqualOptions equal_options = EqualOptions::defaults().nans_equal(true);
            return lit->scalar()->equals(*other.literal()->scalar(), equal_options);
        }

        if (auto ref = field_ref()) {
            return ref->equals(*other.field_ref());
        }

        auto call = CallNotNull(*this);
        auto other_call = CallNotNull(other);

        if (call->function_name != other_call->function_name ||
            call->kernel != other_call->kernel) {
            return false;
        }

        for (size_t i = 0; i < call->arguments.size(); ++i) {
            if (!call->arguments[i].equals(other_call->arguments[i])) {
                return false;
            }
        }

        if (call->options == other_call->options) return true;
        if (call->options && other_call->options) {
            return call->options->equals(*other_call->options);
        }
        return false;
    }

    bool Identical(const Expression &l, const Expression &r) { return l.impl_ == r.impl_; }

    size_t Expression::hash() const {
        if (auto lit = literal()) {
            if (lit->is_scalar()) {
                return lit->scalar()->hash();
            }
            return 0;
        }

        if (auto ref = field_ref()) {
            return ref->hash();
        }

        return CallNotNull(*this)->hash;
    }

    bool Expression::is_bound() const {
        if (type() == nullptr) return false;

        if (const Call *call = this->call()) {
            if (call->kernel == nullptr) return false;

            for (const Expression &arg: call->arguments) {
                if (!arg.is_bound()) return false;
            }
        }

        return true;
    }

    bool Expression::is_scalar_expression() const {
        if (auto lit = literal()) {
            return lit->is_scalar();
        }

        if (field_ref()) return true;

        auto call = CallNotNull(*this);

        for (const Expression &arg: call->arguments) {
            if (!arg.is_scalar_expression()) return false;
        }

        if (call->function) {
            return call->function->kind() == compute::Function::SCALAR;
        }

        // this expression is not bound; make a best guess based on
        // the default function registry
        if (auto function = compute::get_function_registry()
                ->get_function(call->function_name)
                .value_or(nullptr)) {
            return function->kind() == compute::Function::SCALAR;
        }

        // unknown function or other error; conservatively return false
        return false;
    }

    bool Expression::is_null_literal() const {
        if (auto lit = literal()) {
            if (lit->null_count() == lit->length()) {
                return true;
            }
        }

        return false;
    }

    namespace {
        std::optional<compute::NullHandling::type> GetNullHandling(const Expression::Call &call) {
                    DKCHECK_NE(call.function, nullptr);
            if (call.function->kind() == compute::Function::SCALAR) {
                return static_cast<const compute::ScalarKernel *>(call.kernel)->null_handling;
            }
            return std::nullopt;
        }
    }  // namespace

    bool Expression::is_satisfiable() const {
        if (type() == nullptr) return true;
        if (type()->id() != Type::BOOL) return true;

        if (auto lit = literal()) {
            if (lit->null_count() == lit->length()) {
                return false;
            }

            if (lit->is_scalar()) {
                return lit->scalar_as<BooleanScalar>().value;
            }

            return true;
        }

        if (field_ref()) return true;

        auto call = CallNotNull(*this);

        // invert(true_unless_null(x)) is always false or null by definition
        // true_unless_null arises in simplification of inequalities below
        if (call->function_name == "invert") {
            if (auto nested_call = call->arguments[0].call()) {
                if (nested_call->function_name == "true_unless_null") return false;
            }
        }

        if (call->function_name == "and_kleene" || call->function_name == "and") {
            return std::all_of(call->arguments.begin(), call->arguments.end(),
                               [](const Expression &arg) { return arg.is_satisfiable(); });
        }
        if (call->function_name == "or_kleene" || call->function_name == "or") {
            return std::any_of(call->arguments.begin(), call->arguments.end(),
                               [](const Expression &arg) { return arg.is_satisfiable(); });
        }

        return true;
    }

    namespace {

        TypeHolder SmallestTypeFor(const nebula::Datum &value) {
            switch (value.type()->id()) {
                case Type::INT8:
                    return int8();
                case Type::UINT8:
                    return uint8();
                case Type::INT16: {
                    int16_t i16 = value.scalar_as<Int16Scalar>().value;
                    if (i16 <= std::numeric_limits<int8_t>::max() &&
                        i16 >= std::numeric_limits<int8_t>::min()) {
                        return int8();
                    }
                    return int16();
                }
                case Type::UINT16: {
                    uint16_t ui16 = value.scalar_as<UInt16Scalar>().value;
                    if (ui16 <= std::numeric_limits<uint8_t>::max()) {
                        return uint8();
                    }
                    return uint16();
                }
                case Type::INT32: {
                    int32_t i32 = value.scalar_as<Int32Scalar>().value;
                    if (i32 <= std::numeric_limits<int8_t>::max() &&
                        i32 >= std::numeric_limits<int8_t>::min()) {
                        return int8();
                    }
                    if (i32 <= std::numeric_limits<int16_t>::max() &&
                        i32 >= std::numeric_limits<int16_t>::min()) {
                        return int16();
                    }
                    return int32();
                }
                case Type::UINT32: {
                    uint32_t ui32 = value.scalar_as<UInt32Scalar>().value;
                    if (ui32 <= std::numeric_limits<uint8_t>::max()) {
                        return uint8();
                    }
                    if (ui32 <= std::numeric_limits<uint16_t>::max()) {
                        return uint16();
                    }
                    return uint32();
                }
                case Type::INT64: {
                    int64_t i64 = value.scalar_as<Int64Scalar>().value;
                    if (i64 <= std::numeric_limits<int8_t>::max() &&
                        i64 >= std::numeric_limits<int8_t>::min()) {
                        return int8();
                    }
                    if (i64 <= std::numeric_limits<int16_t>::max() &&
                        i64 >= std::numeric_limits<int16_t>::min()) {
                        return int16();
                    }
                    if (i64 <= std::numeric_limits<int32_t>::max() &&
                        i64 >= std::numeric_limits<int32_t>::min()) {
                        return int32();
                    }
                    return int64();
                }
                case Type::UINT64: {
                    uint64_t ui64 = value.scalar_as<UInt64Scalar>().value;
                    if (ui64 <= std::numeric_limits<uint8_t>::max()) {
                        return uint8();
                    }
                    if (ui64 <= std::numeric_limits<uint16_t>::max()) {
                        return uint16();
                    }
                    if (ui64 <= std::numeric_limits<uint32_t>::max()) {
                        return uint32();
                    }
                    return uint64();
                }
                case Type::FP64: {
                    double doub = value.scalar_as<Fp64Scalar>().value;
                    if (!std::isfinite(doub)) {
                        // Special values can be float
                        return float32();
                    }
                    // Test if float representation is the same
                    if (static_cast<double>(static_cast<float>(doub)) == doub) {
                        return float32();
                    }
                    return float64();
                }
                case Type::LARGE_STRING: {
                    if (value.scalar_as<LargeStringScalar>().value->size() <=
                        std::numeric_limits<int32_t>::max()) {
                        return utf8();
                    }
                    return large_utf8();
                }
                case Type::LARGE_BINARY:
                    if (value.scalar_as<LargeBinaryScalar>().value->size() <=
                        std::numeric_limits<int32_t>::max()) {
                        return binary();
                    }
                    return large_binary();
                case Type::TIMESTAMP: {
                    const auto &ts_type = turbo::checked_pointer_cast<TimestampType>(value.type());
                    uint64_t ts = value.scalar_as<TimestampScalar>().value;
                    switch (ts_type->unit()) {
                        case TimeUnit::SECOND:
                            return value.type();
                        case TimeUnit::MILLI:
                            if (ts % 1000 == 0) {
                                return timestamp(TimeUnit::SECOND, ts_type->timezone());
                            }
                            return value.type();
                        case TimeUnit::MICRO:
                            if (ts % 1000000 == 0) {
                                return timestamp(TimeUnit::SECOND, ts_type->timezone());
                            }
                            if (ts % 1000 == 0) {
                                return timestamp(TimeUnit::MILLI, ts_type->timezone());
                            }
                            return value.type();
                        case TimeUnit::NANO:
                            if (ts % 1000000000 == 0) {
                                return timestamp(TimeUnit::SECOND, ts_type->timezone());
                            }
                            if (ts % 1000000 == 0) {
                                return timestamp(TimeUnit::MILLI, ts_type->timezone());
                            }
                            if (ts % 1000 == 0) {
                                return timestamp(TimeUnit::MICRO, ts_type->timezone());
                            }
                            return value.type();
                        default:
                            return value.type();
                    }
                }
                default:
                    return value.type();
            }
        }

        inline std::vector<TypeHolder> GetTypesWithSmallestLiteralRepresentation(
                const std::vector<Expression> &exprs) {
            std::vector<TypeHolder> types(exprs.size());
            for (size_t i = 0; i < exprs.size(); ++i) {
                        DKCHECK(exprs[i].is_bound());
                if (const Datum *literal = exprs[i].literal()) {
                    if (literal->is_scalar()) {
                        types[i] = SmallestTypeFor(*literal);
                    }
                } else {
                    types[i] = exprs[i].type();
                }
            }
            return types;
        }

        // Produce a bound Expression from unbound Call and bound arguments.
        turbo::Result<Expression> BindNonRecursive(Expression::Call call, bool insert_implicit_casts,
                                                   compute::ExecContext *exec_context) {
                    DKCHECK(std::all_of(call.arguments.begin(), call.arguments.end(),
                                              [](const Expression &argument) { return argument.is_bound(); }));

            std::vector<TypeHolder> types = get_types(call.arguments);
            TURBO_MOVE_OR_RAISE(call.function, get_function(call, exec_context));

            auto FinishBind = [&] {
                compute::KernelContext kernel_context(exec_context, call.kernel);
                if (call.kernel->init) {
                    const FunctionOptions *options =
                            call.options ? call.options.get() : call.function->default_options();
                    TURBO_MOVE_OR_RAISE(
                            call.kernel_state,
                            call.kernel->init(&kernel_context, {call.kernel, types, options}));

                    kernel_context.set_state(call.kernel_state.get());
                }

                TURBO_MOVE_OR_RAISE(
                        call.type, call.kernel->signature->out_type().resolve(&kernel_context, types));
                return turbo::OkStatus();
            };

            // First try and bind exactly
            turbo::Result<const Kernel *> maybe_exact_match = call.function->dispatch_exact(types);
            if (maybe_exact_match.ok()) {
                call.kernel = *maybe_exact_match;
                if (FinishBind().ok()) {
                    return Expression(std::move(call));
                }
            }

            if (!insert_implicit_casts) {
                return maybe_exact_match.status();
            }

            // If exact binding fails, and we are allowed to cast, then prefer casting literals
            // first.  Since dispatch_best generally prefers up-casting the best way to do this is
            // first down-cast the literals as much as possible
            types = GetTypesWithSmallestLiteralRepresentation(call.arguments);
            TURBO_MOVE_OR_RAISE(call.kernel, call.function->dispatch_best(&types));

            for (size_t i = 0; i < types.size(); ++i) {
                if (types[i] == call.arguments[i].type()) continue;

                if (const Datum *lit = call.arguments[i].literal()) {
                    TURBO_MOVE_OR_RAISE(Datum new_lit, compute::Cast(*lit, types[i].get_shared_ptr()));
                    call.arguments[i] = literal(std::move(new_lit));
                    continue;
                }

                // construct an implicit cast Expression with which to replace this argument
                Expression::Call implicit_cast;
                implicit_cast.function_name = "cast";
                implicit_cast.arguments = {std::move(call.arguments[i])};

                // TODO(wesm): Use TypeHolder in options
                implicit_cast.options = std::make_shared<compute::CastOptions>(
                        compute::CastOptions::Safe(types[i].get_shared_ptr()));

                TURBO_MOVE_OR_RAISE(
                        call.arguments[i],
                        BindNonRecursive(std::move(implicit_cast),
                                /*insert_implicit_casts=*/false, exec_context));
            }

            TURBO_RETURN_NOT_OK(FinishBind());
            return Expression(std::move(call));
        }

        template<typename TypeOrSchema>
        turbo::Result<Expression> BindImpl(Expression expr, const TypeOrSchema &in,
                                           compute::ExecContext *exec_context) {
            if (exec_context == nullptr) {
                compute::ExecContext exec_context;
                return BindImpl(std::move(expr), in, &exec_context);
            }

            if (expr.literal()) return expr;

            if (const FieldRef *ref = expr.field_ref()) {
                TURBO_MOVE_OR_RAISE(FieldPath path, ref->find_one(in));

                Expression::Parameter param = *expr.parameter();
                param.indices.resize(path.indices().size());
                std::copy(path.indices().begin(), path.indices().end(), param.indices.begin());
                TURBO_MOVE_OR_RAISE(auto field, path.get(in));
                param.type = field->type();
                return Expression{std::move(param)};
            }

            auto call = *CallNotNull(expr);
            for (auto &argument: call.arguments) {
                TURBO_MOVE_OR_RAISE(argument, BindImpl(std::move(argument), in, exec_context));
            }
            return BindNonRecursive(std::move(call),
                    /*insert_implicit_casts=*/true, exec_context);
        }

    }  // namespace

    turbo::Result<Expression> Expression::Bind(const TypeHolder &in,
                                               compute::ExecContext *exec_context) const {
        return BindImpl(*this, *in.type, exec_context);
    }

    turbo::Result<Expression> Expression::Bind(const Schema &in_schema,
                                               compute::ExecContext *exec_context) const {
        return BindImpl(*this, in_schema, exec_context);
    }

    turbo::Result<ExecBatch> make_exec_batch(const Schema &full_schema, const Datum &partial,
                                           Expression guarantee) {
        ExecBatch out;

        if (partial.kind() == Datum::RECORD_BATCH) {
            const auto &partial_batch = *partial.record_batch();
            out.guarantee = std::move(guarantee);
            out.length = partial_batch.num_rows();

            TURBO_MOVE_OR_RAISE(auto known_field_values,
                                extract_known_field_values(out.guarantee));

            for (const auto &field: full_schema.fields()) {
                auto field_ref = FieldRef(field->name());

                // If we know what the value must be from the guarantee, prefer to use that value
                // than the data from the record batch (if it exists at all -- probably it doesn't),
                // because this way it will be a scalar.
                auto known_field_value = known_field_values.map.find(field_ref);
                if (known_field_value != known_field_values.map.end()) {
                    out.values.emplace_back(known_field_value->second);
                    continue;
                }

                TURBO_MOVE_OR_RAISE(auto column, field_ref.get_one_or_none(partial_batch));
                if (column) {
                    if (!column->type()->equals(field->type())) {
                        // Referenced field was present but didn't have the expected type.
                        // This *should* be handled by readers, and will just be an error in the future.
                        TURBO_MOVE_OR_RAISE(
                                auto converted,
                                compute::Cast(column, field->type(), compute::CastOptions::Safe()));
                        column = converted.make_array();
                    }
                    out.values.emplace_back(std::move(column));
                } else {
                    out.values.emplace_back(MakeNullScalar(field->type()));
                }
            }
            return out;
        }

        // wasteful but useful for testing:
        if (partial.type()->id() == Type::STRUCT) {
            if (partial.is_array()) {
                TURBO_MOVE_OR_RAISE(auto partial_batch,
                                    RecordBatch::FromStructArray(partial.make_array()));

                return make_exec_batch(full_schema, partial_batch, std::move(guarantee));
            }

            if (partial.is_scalar()) {
                TURBO_MOVE_OR_RAISE(auto partial_array,
                                    MakeArrayFromScalar(*partial.scalar(), 1));
                TURBO_MOVE_OR_RAISE(
                        auto out, make_exec_batch(full_schema, partial_array, std::move(guarantee)));

                for (Datum &value: out.values) {
                    if (value.is_scalar()) continue;
                    TURBO_MOVE_OR_RAISE(value, value.make_array()->get_scalar(0));
                }
                return out;
            }
        }

        return turbo::unimplemented_error("make_exec_batch from ", PrintDatum(partial));
    }

    turbo::Result<Datum> execute_scalar_expression(const Expression &expr, const Schema &full_schema,
                                                 const Datum &partial_input,
                                                 compute::ExecContext *exec_context) {
        TURBO_MOVE_OR_RAISE(auto input, make_exec_batch(full_schema, partial_input));
        return execute_scalar_expression(expr, input, exec_context);
    }

    turbo::Result<Datum> execute_scalar_expression(const Expression &expr, const ExecBatch &input,
                                                 compute::ExecContext *exec_context) {
        if (exec_context == nullptr) {
            compute::ExecContext exec_context;
            return execute_scalar_expression(expr, input, &exec_context);
        }

        if (!expr.is_bound()) {
            return turbo::invalid_argument_error("Cannot execute unbound expression.");
        }

        if (!expr.is_scalar_expression()) {
            return turbo::invalid_argument_error(
                    "execute_scalar_expression cannot execute non-scalar expression ", expr.to_string());
        }

        if (auto lit = expr.literal()) return *lit;

        if (auto param = expr.parameter()) {
            if (param->type.id() == Type::NA) {
                return MakeNullScalar(null());
            }

            Datum field = input[param->indices[0]];
            if (param->indices.size() > 1) {
                std::vector<int> indices(param->indices.begin() + 1, param->indices.end());
                compute::StructFieldOptions options(std::move(indices));
                TURBO_MOVE_OR_RAISE(
                        field, compute::call_function("struct_field", {std::move(field)}, &options));
            }
            if (!field.type()->equals(*param->type.type)) {
                return turbo::invalid_argument_error("Referenced field ", expr.to_string(), " was ",
                                                     field.type()->to_string(), " but should have been ",
                                                     param->type.to_string());
            }

            return field;
        }

        auto call = CallNotNull(expr);

        std::vector<Datum> arguments(call->arguments.size());

        bool all_scalar = true;
        for (size_t i = 0; i < arguments.size(); ++i) {
            TURBO_MOVE_OR_RAISE(
                    arguments[i], execute_scalar_expression(call->arguments[i], input, exec_context));
            all_scalar &= arguments[i].is_scalar();
        }

        int64_t input_length;
        if (!arguments.empty() && all_scalar) {
            // all inputs are scalar, so use a 1-long batch to avoid
            // computing input.length equivalent outputs
            input_length = 1;
        } else {
            input_length = input.length;
        }

        auto executor = compute::detail::KernelExecutor::MakeScalar();

        compute::KernelContext kernel_context(exec_context, call->kernel);
        kernel_context.set_state(call->kernel_state.get());

        const Kernel *kernel = call->kernel;
        std::vector<TypeHolder> types = get_types(arguments);
        auto options = call->options.get();
        TURBO_RETURN_NOT_OK(executor->init(&kernel_context, {kernel, types, options}));

        compute::detail::DatumAccumulator listener;
        TURBO_RETURN_NOT_OK(
                executor->execute(ExecBatch(std::move(arguments), input_length), &listener));
        const auto out = executor->wrap_results(arguments, listener.values());
#ifndef NDEBUG
                KCHECK_OK(executor->CheckResultType(out, call->function_name.c_str()));
#endif
        return out;
    }

    namespace {

        std::array<std::pair<const Expression &, const Expression &>, 2>
        ArgumentsAndFlippedArguments(const Expression::Call &call) {
                    DKCHECK_EQ(call.arguments.size(), 2);
            return {std::pair<const Expression &, const Expression &>{call.arguments[0],
                                                                      call.arguments[1]},
                    std::pair<const Expression &, const Expression &>{call.arguments[1],
                                                                      call.arguments[0]}};
        }

    }  // namespace

    std::vector<FieldRef> fields_in_expression(const Expression &expr) {
        if (expr.literal()) return {};

        if (auto ref = expr.field_ref()) {
            return {*ref};
        }

        std::vector<FieldRef> fields;
        for (const Expression &arg: CallNotNull(expr)->arguments) {
            auto argument_fields = fields_in_expression(arg);
            std::move(argument_fields.begin(), argument_fields.end(), std::back_inserter(fields));
        }
        return fields;
    }

    bool expression_has_field_refs(const Expression &expr) {
        if (expr.literal()) return false;

        if (expr.field_ref()) return true;

        for (const Expression &arg: CallNotNull(expr)->arguments) {
            if (expression_has_field_refs(arg)) return true;
        }
        return false;
    }

    turbo::Result<Expression> fold_constants(Expression expr) {
        if (!expr.is_bound()) {
            return turbo::invalid_argument_error("Cannot fold constants in unbound expression.");
        }

        return ModifyExpression(
                std::move(expr), [](Expression expr) { return expr; },
                [](Expression expr, ...) -> turbo::Result<Expression> {
                    auto call = CallNotNull(expr);
                    if (!call->function->is_pure()) return expr;

                    if (std::all_of(call->arguments.begin(), call->arguments.end(),
                                    [](const Expression &argument) { return argument.literal(); })) {
                        // all arguments are literal; we can evaluate this subexpression *now*
                        static const ExecBatch ignored_input = ExecBatch({}, 1);
                        TURBO_MOVE_OR_RAISE(Datum constant,
                                            execute_scalar_expression(expr, ignored_input));

                        return literal(std::move(constant));
                    }

                    // XXX the following should probably be in a registry of passes instead
                    // of inline

                    if (GetNullHandling(*call) == compute::NullHandling::INTERSECTION) {
                        // kernels which always produce intersected validity can be resolved
                        // to null *now* if any of their inputs is a null literal
                        for (const Expression &argument: call->arguments) {
                            if (argument.is_null_literal()) {
                                if (argument.type()->equals(*call->type.type)) {
                                    return argument;
                                } else {
                                    return literal(MakeNullScalar(call->type.get_shared_ptr()));
                                }
                            }
                        }
                    }

                    if (call->function_name == "and_kleene") {
                        for (auto args: ArgumentsAndFlippedArguments(*call)) {
                            // true and x == x
                            if (args.first == literal(true)) return args.second;

                            // false and x == false
                            if (args.first == literal(false)) return args.first;

                            // x and x == x
                            if (args.first == args.second) return args.first;
                        }
                        return expr;
                    }

                    if (call->function_name == "or_kleene") {
                        for (auto args: ArgumentsAndFlippedArguments(*call)) {
                            // false or x == x
                            if (args.first == literal(false)) return args.second;

                            // true or x == true
                            if (args.first == literal(true)) return args.first;

                            // x or x == x
                            if (args.first == args.second) return args.first;
                        }
                        return expr;
                    }

                    return expr;
                });
    }

    namespace {

        std::vector<Expression> GuaranteeConjunctionMembers(
                const Expression &guaranteed_true_predicate) {
            auto guarantee = guaranteed_true_predicate.call();
            if (!guarantee || guarantee->function_name != "and_kleene") {
                return {guaranteed_true_predicate};
            }
            return FlattenedAssociativeChain(guaranteed_true_predicate).fringe;
        }

        /// \brief Extract an equality from an expression.
        ///
        /// Recognizes expressions of the form:
        /// equal(a, 2)
        /// is_null(a)
        std::optional<std::pair<FieldRef, Datum>> ExtractOneFieldValue(
                const Expression &guarantee) {
            auto call = guarantee.call();
            if (!call) return std::nullopt;

            // search for an equality conditions between a field and a literal
            if (call->function_name == "equal") {
                auto ref = call->arguments[0].field_ref();
                if (!ref) return std::nullopt;

                auto lit = call->arguments[1].literal();
                if (!lit) return std::nullopt;

                return std::make_pair(*ref, *lit);
            }

            // ... or a known null field
            if (call->function_name == "is_null") {
                auto ref = call->arguments[0].field_ref();
                if (!ref) return std::nullopt;

                return std::make_pair(*ref, Datum(std::make_shared<NullScalar>()));
            }

            return std::nullopt;
        }

        // Conjunction members which are represented in known_values are erased from
        // conjunction_members
        turbo::Status extract_known_field_values(std::vector<Expression> *conjunction_members,
                                              KnownFieldValues *known_values) {
            // filter out consumed conjunction members, leaving only unconsumed
            *conjunction_members = nebula::internal::FilterVector(
                    std::move(*conjunction_members),
                    [known_values](const Expression &guarantee) -> bool {
                        if (auto known_value = ExtractOneFieldValue(guarantee)) {
                            known_values->map.insert(std::move(*known_value));
                            return false;
                        }
                        return true;
                    });

            return turbo::OkStatus();
        }

    }  // namespace

    turbo::Result<KnownFieldValues> extract_known_field_values(
            const Expression &guaranteed_true_predicate) {
        KnownFieldValues known_values;
        auto conjunction_members = GuaranteeConjunctionMembers(guaranteed_true_predicate);
        TURBO_RETURN_NOT_OK(extract_known_field_values(&conjunction_members, &known_values));
        return known_values;
    }

    turbo::Result<Expression> replace_fields_with_known_values(const KnownFieldValues &known_values,
                                                           Expression expr) {
        if (!expr.is_bound()) {
            return turbo::invalid_argument_error(
                    "replace_fields_with_known_values called on an unbound Expression");
        }

        return ModifyExpression(
                std::move(expr),
                [&known_values](Expression expr) -> turbo::Result<Expression> {
                    if (auto ref = expr.field_ref()) {
                        auto it = known_values.map.find(*ref);
                        if (it != known_values.map.end()) {
                            Datum lit = it->second;
                            if (lit.type()->equals(*expr.type())) return literal(std::move(lit));
                            // type mismatch, try casting the known value to the correct type

                            if (expr.type()->id() == Type::DICTIONARY &&
                                lit.type()->id() != Type::DICTIONARY) {
                                // the known value must be dictionary encoded

                                const auto &dict_type = turbo::checked_cast<const DictionaryType &>(*expr.type());
                                if (!lit.type()->equals(dict_type.get_value_type())) {
                                    TURBO_MOVE_OR_RAISE(lit, compute::Cast(lit, dict_type.get_value_type()));
                                }

                                if (lit.is_scalar()) {
                                    TURBO_MOVE_OR_RAISE(auto dictionary,
                                                        MakeArrayFromScalar(*lit.scalar(), 1));

                                    lit = Datum{DictionaryScalar::create(MakeScalar<int32_t>(0),
                                                                       std::move(dictionary))};
                                }
                            }

                            TURBO_MOVE_OR_RAISE(lit, compute::Cast(lit, expr.type()->get_shared_ptr()));
                            return literal(std::move(lit));
                        }
                    }
                    return expr;
                },
                [](Expression expr, ...) { return expr; });
    }

    namespace {

        bool IsBinaryAssociativeCommutative(const Expression::Call &call) {
            static std::unordered_set<std::string> binary_associative_commutative{
                    "and", "or", "and_kleene", "or_kleene", "xor",
                    "multiply", "add", "multiply_checked", "add_checked"};

            auto it = binary_associative_commutative.find(call.function_name);
            return it != binary_associative_commutative.end();
        }

        turbo::Result<Expression> HandleInconsistentTypes(Expression::Call call,
                                                          compute::ExecContext *exec_context) {
            // ARROW-18334: due to reordering of arguments, the call may have
            // inconsistent argument types. For example, the call's kernel may
            // correspond to `timestamp + duration` but the arguments happen to
            // be `duration, timestamp`. The addition itself is still commutative,
            // but the mismatch in declared argument types is potentially problematic
            // if we ever start using the Expression::Call::kernel field more than
            // we do currently. Check and rebind if necessary.
            //
            // The more correct fix for this problem is to ensure that all kernels of
            // functions which are commutative be commutative as well, which would
            // obviate rebinding like this. In the context of ARROW-18334, this
            // would require rewriting KernelSignature so that a single kernel can
            // handle both `timestamp + duration` and `duration + timestamp`.
            if (call.kernel->signature->matches_inputs(get_types(call.arguments))) {
                return Expression(std::move(call));
            }
            return BindNonRecursive(std::move(call), /*insert_implicit_casts=*/false, exec_context);
        }

    }  // namespace

    turbo::Result<Expression> canonicalize(Expression expr, compute::ExecContext *exec_context) {
        if (!expr.is_bound()) {
            return turbo::invalid_argument_error("Cannot canonicalize an unbound expression.");
        }

        if (exec_context == nullptr) {
            compute::ExecContext exec_context;
            return canonicalize(std::move(expr), &exec_context);
        }

        // If potentially reconstructing more deeply than a call's immediate arguments
        // (for example, when reorganizing an associative chain), add expressions to this set to
        // avoid unnecessary work
        struct {
            std::unordered_set<Expression, Expression::Hash> set_;

            bool operator()(const Expression &expr) const {
                return set_.find(expr) != set_.end();
            }

            void Add(std::vector<Expression> exprs) {
                std::move(exprs.begin(), exprs.end(), std::inserter(set_, set_.end()));
            }
        } AlreadyCanonicalized;

        return ModifyExpression(
                std::move(expr),
                [&AlreadyCanonicalized, exec_context](Expression expr) -> turbo::Result<Expression> {
                    auto call = expr.call();
                    if (!call) return expr;
                    if (!call->function->is_pure()) return expr;

                    if (AlreadyCanonicalized(expr)) return expr;

                    if (IsBinaryAssociativeCommutative(*call)) {
                        struct {
                            int Priority(const Expression &operand) const {
                                // order literals first, starting with nulls
                                if (operand.is_null_literal()) return 0;
                                if (operand.literal()) return 1;
                                return 2;
                            }

                            bool operator()(const Expression &l, const Expression &r) const {
                                return Priority(l) < Priority(r);
                            }
                        } CanonicalOrdering;

                        FlattenedAssociativeChain chain(expr);

                        if (chain.was_left_folded &&
                            std::is_sorted(chain.fringe.begin(), chain.fringe.end(),
                                           CanonicalOrdering)) {
                            // fast path for expressions which happen to have arrived in an
                            // already-canonical form
                            AlreadyCanonicalized.Add(std::move(chain.exprs));
                            return expr;
                        }

                        std::stable_sort(chain.fringe.begin(), chain.fringe.end(), CanonicalOrdering);

                        // fold the chain back up
                        Expression folded = std::move(chain.fringe.front());

                        for (auto it = chain.fringe.begin() + 1; it != chain.fringe.end(); ++it) {
                            auto canonicalized_call = *call;
                            canonicalized_call.arguments = {std::move(folded), std::move(*it)};
                            TURBO_MOVE_OR_RAISE(
                                    folded,
                                    HandleInconsistentTypes(std::move(canonicalized_call), exec_context));
                            AlreadyCanonicalized.Add({expr});
                        }
                        return folded;
                    }

                    if (auto cmp = Comparison::Get(call->function_name)) {
                        if (call->arguments[0].literal() && !call->arguments[1].literal()) {
                            // ensure that literals are on comparisons' RHS
                            auto flipped_call = *call;

                            std::swap(flipped_call.arguments[0], flipped_call.arguments[1]);
                            flipped_call.function_name =
                                    Comparison::GetName(Comparison::GetFlipped(*cmp));

                            return BindNonRecursive(flipped_call,
                                    /*insert_implicit_casts=*/false, exec_context);
                        }
                    }

                    return expr;
                },
                [](Expression expr, ...) { return expr; });
    }

    namespace {

        // An inequality comparison which a target Expression is known to satisfy. If nullable,
        // the target may evaluate to null in addition to values satisfying the comparison.
        struct Inequality {
            // The inequality type
            Comparison::type cmp;
            // The LHS of the inequality
            const FieldRef &target;
            // The RHS of the inequality
            const Datum &bound;
            // Whether target can be null
            bool nullable;

            // Extract an Inequality if possible, derived from "less",
            // "greater", "less_equal", and "greater_equal" expressions,
            // possibly disjuncted with an "is_null" Expression.
            // cmp(a, 2)
            // cmp(a, 2) or is_null(a)
            static std::optional<Inequality> ExtractOne(const Expression &guarantee) {
                auto call = guarantee.call();
                if (!call) return std::nullopt;

                if (call->function_name == "or_kleene") {
                    // expect the LHS to be a usable field inequality
                    auto out = ExtractOneFromComparison(call->arguments[0]);
                    if (!out) return std::nullopt;

                    // expect the RHS to be an is_null expression
                    auto call_rhs = call->arguments[1].call();
                    if (!call_rhs) return std::nullopt;
                    if (call_rhs->function_name != "is_null") return std::nullopt;

                    // ... and that it references the same target
                    auto target = call_rhs->arguments[0].field_ref();
                    if (!target) return std::nullopt;
                    if (*target != out->target) return std::nullopt;

                    out->nullable = true;
                    return out;
                }

                // fall back to a simple comparison with no "is_null"
                return ExtractOneFromComparison(guarantee);
            }

            static std::optional<Inequality> ExtractOneFromComparison(const Expression &guarantee) {
                auto call = guarantee.call();
                if (!call) return std::nullopt;

                if (auto cmp = Comparison::Get(call->function_name)) {
                    // not_equal comparisons are not very usable as guarantees
                    if (*cmp == Comparison::NOT_EQUAL) return std::nullopt;

                    auto target = call->arguments[0].field_ref();
                    if (!target) return std::nullopt;

                    auto bound = call->arguments[1].literal();
                    if (!bound) return std::nullopt;
                    if (!bound->is_scalar()) return std::nullopt;

                    return Inequality{*cmp, /*target=*/*target, *bound, /*nullable=*/false};
                }

                return std::nullopt;
            }

            /// The given expression simplifies to `value` if the inequality
            /// target is not nullable. Otherwise, it simplifies to either a
            /// call to true_unless_null or !true_unless_null.
            turbo::Result<Expression> simplified_to(const Expression &bound_target, bool value) const {
                if (!nullable) return literal(value);

                ExecContext exec_context;

                // Data may be null, so comparison will yield `value` - or null IFF the data was null
                //
                // true_unless_null is cheap; it purely reuses the validity bitmap for the values
                // buffer. Inversion is less cheap but we expect that term never to be evaluated
                // since invert(true_unless_null(x)) is not satisfiable.
                Expression::Call call;
                call.function_name = "true_unless_null";
                call.arguments = {bound_target};
                TURBO_MOVE_OR_RAISE(
                        auto true_unless_null,
                        BindNonRecursive(std::move(call),
                                /*insert_implicit_casts=*/false, &exec_context));
                if (value) return true_unless_null;

                Expression::Call invert;
                invert.function_name = "invert";
                invert.arguments = {std::move(true_unless_null)};
                return BindNonRecursive(std::move(invert),
                        /*insert_implicit_casts=*/false, &exec_context);
            }

            /// \brief Simplify the given expression given this inequality as a guarantee.
            turbo::Result<Expression> Simplify(Expression expr) {
                const auto &guarantee = *this;

                auto call = expr.call();
                if (!call) return expr;

                if (call->function_name == "is_valid" || call->function_name == "is_null") {
                    if (guarantee.nullable) return expr;
                    const auto &lhs = Comparison::StripOrderPreservingCasts(call->arguments[0]);
                    if (!lhs.field_ref()) return expr;
                    if (*lhs.field_ref() != guarantee.target) return expr;

                    return call->function_name == "is_valid" ? literal(true) : literal(false);
                }

                auto cmp = Comparison::Get(expr);
                if (!cmp) return expr;

                auto rhs = call->arguments[1].literal();
                if (!rhs) return expr;
                if (!rhs->is_scalar()) return expr;

                const auto &lhs = Comparison::StripOrderPreservingCasts(call->arguments[0]);
                if (!lhs.field_ref()) return expr;
                if (*lhs.field_ref() != guarantee.target) return expr;

                // Whether the RHS of the expression is EQUAL, LESS, or GREATER than the
                // RHS of the guarantee. N.B. Comparison::type is a bitmask
                TURBO_MOVE_OR_RAISE(const Comparison::type cmp_rhs_bound,
                                    Comparison::execute(*rhs, guarantee.bound));
                        DKCHECK_NE(cmp_rhs_bound, Comparison::NA);

                if (cmp_rhs_bound == Comparison::EQUAL) {
                    // RHS of filter is equal to RHS of guarantee

                    if ((*cmp & guarantee.cmp) == guarantee.cmp) {
                        // guarantee is a subset of filter, so all data will be included
                        // x > 1, x >= 1, x != 1 guaranteed by x > 1
                        return simplified_to(lhs, true);
                    }

                    if ((*cmp & guarantee.cmp) == 0) {
                        // guarantee disjoint with filter, so all data will be excluded
                        // x > 1, x >= 1 unsatisfiable if x == 1
                        return simplified_to(lhs, false);
                    }

                    return expr;
                }

                if (guarantee.cmp & cmp_rhs_bound) {
                    // We guarantee (x (?) N) and are trying to simplify (x (?) M).  We know
                    // either M < N or M > N (i.e. cmp_rhs_bound is either LESS or GREATER).

                    // If M > N, then if the guarantee is (x > N), (x >= N), or (x != N)
                    // (i.e. guarantee.cmp & cmp_rhs_bound), we cannot do anything with the
                    // guarantee, and bail out here.

                    // For example, take M = 5, N = 3. Then cmp_rhs_bound = GREATER.
                    // x > 3, x >= 3, x != 3 implies nothing about x < 5, x <= 5, x > 5,
                    // x >= 5, x != 5 and we bail out here.
                    // x < 3, x <= 3 could simplify (some of) those expressions.
                    return expr;
                }

                if (*cmp & Comparison::GetFlipped(cmp_rhs_bound)) {
                    // x > 1, x >= 1, x != 1 guaranteed by x >= 3
                    // (where `guarantee.cmp` is GREATER_EQUAL, `cmp_rhs_bound` is LESS)
                    return simplified_to(lhs, true);
                } else {
                    // x < 1, x <= 1, x == 1 unsatisfiable if x >= 3
                    return simplified_to(lhs, false);
                }
            }
        };

        /// \brief Simplify an expression given a guarantee, if the guarantee
        ///   is is_valid().
        turbo::Result<Expression> SimplifyIsValidGuarantee(Expression expr,
                                                           const Expression::Call &guarantee) {
            if (guarantee.function_name != "is_valid") return expr;

            return ModifyExpression(
                    std::move(expr), [](Expression expr) { return expr; },
                    [&](Expression expr, ...) -> turbo::Result<Expression> {
                        auto call = expr.call();
                        if (!call) return expr;

                        if (call->arguments[0] != guarantee.arguments[0]) return expr;

                        if (call->function_name == "is_valid") return literal(true);

                        if (call->function_name == "true_unless_null") return literal(true);

                        if (call->function_name == "is_null") return literal(false);

                        return expr;
                    });
        }

    }  // namespace

    turbo::Result<Expression> simplify_with_guarantee(Expression expr,
                                                    const Expression &guaranteed_true_predicate) {
        KnownFieldValues known_values;
        auto conjunction_members = GuaranteeConjunctionMembers(guaranteed_true_predicate);

        TURBO_RETURN_NOT_OK(extract_known_field_values(&conjunction_members, &known_values));

        TURBO_MOVE_OR_RAISE(expr,
                            replace_fields_with_known_values(known_values, std::move(expr)));

        auto CanonicalizeAndFoldConstants = [&expr] {
            TURBO_MOVE_OR_RAISE(expr, canonicalize(std::move(expr)));
            TURBO_MOVE_OR_RAISE(expr, fold_constants(std::move(expr)));
            return turbo::OkStatus();
        };
        TURBO_RETURN_NOT_OK(CanonicalizeAndFoldConstants());

        for (const auto &guarantee: conjunction_members) {
            if (!guarantee.call()) continue;

            if (auto inequality = Inequality::ExtractOne(guarantee)) {
                TURBO_MOVE_OR_RAISE(auto simplified,
                                    ModifyExpression(
                                            std::move(expr), [](Expression expr) { return expr; },
                                            [&](Expression expr, ...) -> turbo::Result<Expression> {
                                                return inequality->Simplify(std::move(expr));
                                            }));

                if (Identical(simplified, expr)) continue;

                expr = std::move(simplified);
                TURBO_RETURN_NOT_OK(CanonicalizeAndFoldConstants());
            }

            if (guarantee.call()->function_name == "is_valid") {
                TURBO_MOVE_OR_RAISE(
                        auto simplified,
                        SimplifyIsValidGuarantee(std::move(expr), *CallNotNull(guarantee)));

                if (Identical(simplified, expr)) continue;

                expr = std::move(simplified);
                TURBO_RETURN_NOT_OK(CanonicalizeAndFoldConstants());
            }
        }

        return expr;
    }

    turbo::Result<Expression> remove_named_refs(Expression src) {
        if (!src.is_bound()) {
            return turbo::invalid_argument_error("remove_named_refs called on unbound expression");
        }
        return ModifyExpression(
                std::move(src),
                /*pre=*/
                [](Expression expr) {
                    const Expression::Parameter *param = expr.parameter();
                    if (param && !param->ref.is_field_path()) {
                        FieldPath ref_as_path(
                                std::vector<int>(param->indices.begin(), param->indices.end()));
                        return Expression(
                                Expression::Parameter{std::move(ref_as_path), param->type, param->indices});
                    }

                    return expr;
                },
                /*post_call=*/[](Expression expr, ...) { return expr; });
    }

    // Serialization is accomplished by converting expressions to KeyValueMetadata and storing
    // this in the schema of a RecordBatch. Embedded arrays and scalars are stored in its
    // columns. Finally, the RecordBatch is written to an IPC file.
    turbo::Result<std::shared_ptr<Buffer>> serialize(const Expression &expr) {
        struct {
            std::shared_ptr<KeyValueMetadata> metadata_ = std::make_shared<KeyValueMetadata>();
            ArrayVector columns_;

            turbo::Result<std::string> AddScalar(const Scalar &scalar) {
                auto ret = columns_.size();
                TURBO_MOVE_OR_RAISE(auto array, MakeArrayFromScalar(scalar, 1));
                columns_.push_back(std::move(array));
                return ToChars(ret);
            }

            turbo::Status VisitFieldRef(const FieldRef &ref) {
                if (ref.nested_refs()) {
                    metadata_->append("nested_field_ref", ToChars(ref.nested_refs()->size()));
                    for (const auto &child: *ref.nested_refs()) {
                        TURBO_RETURN_NOT_OK(VisitFieldRef(child));
                    }
                    return turbo::OkStatus();
                }
                if (!ref.name()) {
                    return turbo::unimplemented_error("Serialization of non-name field_refs");
                }
                metadata_->append("field_ref", *ref.name());
                return turbo::OkStatus();
            }

            turbo::Status Visit(const Expression &expr) {
                if (auto lit = expr.literal()) {
                    if (!lit->is_scalar()) {
                        return turbo::unimplemented_error("Serialization of non-scalar literals");
                    }
                    TURBO_MOVE_OR_RAISE(auto value, AddScalar(*lit->scalar()));
                    metadata_->append("literal", std::move(value));
                    return turbo::OkStatus();
                }

                if (auto ref = expr.field_ref()) {
                    return VisitFieldRef(*ref);
                }

                auto call = CallNotNull(expr);
                metadata_->append("call", call->function_name);

                for (const auto &argument: call->arguments) {
                    TURBO_RETURN_NOT_OK(Visit(argument));
                }

                if (call->options) {
                    TURBO_MOVE_OR_RAISE(auto options_scalar,
                                        internal::FunctionOptionsToStructScalar(*call->options));
                    TURBO_MOVE_OR_RAISE(auto value, AddScalar(*options_scalar));
                    metadata_->append("options", std::move(value));
                }

                metadata_->append("end", call->function_name);
                return turbo::OkStatus();
            }

            turbo::Result<std::shared_ptr<RecordBatch>> operator()(const Expression &expr) {
                TURBO_RETURN_NOT_OK(Visit(expr));
                FieldVector fields(columns_.size());
                for (size_t i = 0; i < fields.size(); ++i) {
                    fields[i] = field("", columns_[i]->type());
                }
                return RecordBatch::create(schema(std::move(fields), std::move(metadata_)), 1,
                                         std::move(columns_));
            }
        } to_record_batch;

        TURBO_MOVE_OR_RAISE(auto batch, to_record_batch(expr));
        TURBO_MOVE_OR_RAISE(auto stream, io::BufferOutputStream::create());
        TURBO_MOVE_OR_RAISE(auto writer, ipc::make_file_writer(stream, batch->schema()));
        TURBO_RETURN_NOT_OK(writer->write_record_batch(*batch));
        TURBO_RETURN_NOT_OK(writer->close());
        return stream->finish();
    }

    turbo::Result<Expression> deserialize(std::shared_ptr<Buffer> buffer) {
        io::BufferReader stream(std::move(buffer));
        TURBO_MOVE_OR_RAISE(auto reader, ipc::RecordBatchFileReader::open(&stream));
        TURBO_MOVE_OR_RAISE(auto batch, reader->read_record_batch(0));
        if (batch->schema()->metadata() == nullptr) {
            return turbo::invalid_argument_error("serialized Expression's batch repr had null metadata");
        }
        if (batch->num_rows() != 1) {
            return turbo::invalid_argument_error(
                    "serialized Expression's batch repr was not a single row - had ",
                    batch->num_rows());
        }

        struct FromRecordBatch {
            const RecordBatch &batch_;
            int index_;

            const KeyValueMetadata &metadata() { return *batch_.schema()->metadata(); }

            bool ParseInteger(const std::string &s, int32_t *value) {
                return ::nebula::internal::ParseValue<Int32Type>(s.data(), s.length(), value);
            }

            turbo::Result<std::shared_ptr<Scalar>> get_scalar(const std::string &i) {
                int32_t column_index;
                if (!ParseInteger(i, &column_index)) {
                    return turbo::invalid_argument_error("Couldn't parse column_index");
                }
                if (column_index >= batch_.num_columns()) {
                    return turbo::invalid_argument_error("column_index out of bounds");
                }
                return batch_.column(column_index)->get_scalar(0);
            }

            turbo::Result<Expression> get_one() {
                if (index_ >= metadata().size()) {
                    return turbo::invalid_argument_error("unterminated serialized Expression");
                }

                const std::string &key = metadata().key(index_);
                const std::string &value = metadata().value(index_);
                ++index_;

                if (key == "literal") {
                    TURBO_MOVE_OR_RAISE(auto scalar, get_scalar(value));
                    return literal(std::move(scalar));
                }

                if (key == "nested_field_ref") {
                    int32_t size;
                    if (!ParseInteger(value, &size)) {
                        return turbo::invalid_argument_error("Couldn't parse nested field ref length");
                    }
                    if (size <= 0) {
                        return turbo::invalid_argument_error("nested field ref length must be > 0");
                    }
                    std::vector<FieldRef> nested;
                    nested.reserve(size);
                    while (size-- > 0) {
                        TURBO_MOVE_OR_RAISE(auto ref, get_one());
                        if (!ref.field_ref()) {
                            return turbo::invalid_argument_error("invalid nested field ref");
                        }
                        nested.push_back(*ref.field_ref());
                    }
                    return field_ref(FieldRef(std::move(nested)));
                }

                if (key == "field_ref") {
                    return field_ref(value);
                }

                if (key != "call") {
                    return turbo::invalid_argument_error("Unrecognized serialized Expression key ", key);
                }

                std::vector<Expression> arguments;
                while (metadata().key(index_) != "end") {
                    if (metadata().key(index_) == "options") {
                        TURBO_MOVE_OR_RAISE(auto options_scalar, get_scalar(metadata().value(index_)));
                        std::shared_ptr<compute::FunctionOptions> options;
                        if (options_scalar) {
                            TURBO_MOVE_OR_RAISE(
                                    options, internal::FunctionOptionsFromStructScalar(
                                    turbo::checked_cast<const StructScalar &>(*options_scalar)));
                        }
                        auto expr = call(value, std::move(arguments), std::move(options));
                        index_ += 2;
                        return expr;
                    }

                    TURBO_MOVE_OR_RAISE(auto argument, get_one());
                    arguments.push_back(std::move(argument));
                }

                ++index_;
                return call(value, std::move(arguments));
            }
        };

        return FromRecordBatch{*batch, 0}.get_one();
    }

    Expression project(std::vector<Expression> values, std::vector<std::string> names) {
        return call("make_struct", std::move(values),
                    compute::MakeStructOptions{std::move(names)});
    }

    Expression equal(Expression lhs, Expression rhs) {
        return call("equal", {std::move(lhs), std::move(rhs)});
    }

    Expression not_equal(Expression lhs, Expression rhs) {
        return call("not_equal", {std::move(lhs), std::move(rhs)});
    }

    Expression less(Expression lhs, Expression rhs) {
        return call("less", {std::move(lhs), std::move(rhs)});
    }

    Expression less_equal(Expression lhs, Expression rhs) {
        return call("less_equal", {std::move(lhs), std::move(rhs)});
    }

    Expression greater(Expression lhs, Expression rhs) {
        return call("greater", {std::move(lhs), std::move(rhs)});
    }

    Expression greater_equal(Expression lhs, Expression rhs) {
        return call("greater_equal", {std::move(lhs), std::move(rhs)});
    }

    Expression is_null(Expression lhs, bool nan_is_null) {
        return call("is_null", {std::move(lhs)}, compute::NullOptions(std::move(nan_is_null)));
    }

    Expression is_valid(Expression lhs) { return call("is_valid", {std::move(lhs)}); }

    Expression and_(Expression lhs, Expression rhs) {
        return call("and_kleene", {std::move(lhs), std::move(rhs)});
    }

    Expression and_(const std::vector<Expression> &operands) {
        if (operands.empty()) return literal(true);

        Expression folded = operands.front();
        for (auto it = operands.begin() + 1; it != operands.end(); ++it) {
            folded = and_(std::move(folded), *it);
        }
        return folded;
    }

    Expression or_(Expression lhs, Expression rhs) {
        return call("or_kleene", {std::move(lhs), std::move(rhs)});
    }

    Expression or_(const std::vector<Expression> &operands) {
        if (operands.empty()) return literal(false);

        Expression folded = operands.front();
        for (auto it = operands.begin() + 1; it != operands.end(); ++it) {
            folded = or_(std::move(folded), *it);
        }
        return folded;
    }

    Expression not_(Expression operand) { return call("invert", {std::move(operand)}); }

}  // namespace nebula::compute
